package org.sg20.labelpropagation.lpaiter;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.sg20.utils.Edge;
import org.sg20.utils.LpaDotVal;
import org.sg20.utils.ParseText;


import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashSet;
import java.util.Set;

public class LpaIterMapper extends Mapper<Text, Text,Text, Text> {

    private Set<String> fixedDots; //初始点的标签不会变化
    private Text outKey,outVal;

    @Override
    protected void setup(Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        super.setup(context);
        this.fixedDots=new HashSet<>();
        this.outKey=new Text();
        this.outVal=new Text();
        URI[] cacheFiles = context.getCacheFiles();
        if (cacheFiles != null && cacheFiles.length > 0) {
            /*
                读入初始的,已经带有标签的点集
                格式为"人名\t 标签"
             */
            {
                Path cache = new Path(cacheFiles[0]);
                String line;
                String[] tokens;
                FileSystem fs = FileSystem.get(context.getConfiguration());
                InputStreamReader is = new InputStreamReader(fs.open(cache));
                BufferedReader labelReader = new BufferedReader(is);
                try
                {
                    {
                        while ((line = labelReader.readLine()) != null)
                        {
                            String name = line.split("\t", 2)[0];
                            fixedDots.add(name);
                        }
                    }
                }finally{
                    labelReader.close();
                }
            }
        }
    }

    @Override
    protected void map(Text key, Text value, Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        /*
         *name\t[pr1,pr2,...] [name1,weight1|name2,weight2|...]
         * 将name节点的标签 概率转移到name1，name2..节点上
         * 输出类型1:标签,概率
         * 输出类型2:链接列表
         */
        String name=key.toString();
        String[] temp=value.toString().split(" ");
        LpaDotVal dotVal= ParseText.parseLPADotVal(temp[0]);
        Edge[] edges= ParseText.parseEdges(temp[1]);
        for(Edge edge:edges){
            if(fixedDots.contains(edge.getDestName())) continue; //fixedDots里的点的标签不能被改变
            outKey.set(edge.getDestName());
            double weight=edge.getWeight();
            for(int i=0;i<dotVal.probabilities.length;i++){
                double newProb=dotVal.probabilities[i]*weight;
                if(newProb!=0) {
                    outVal.set(String.format("%d,%f",i,newProb)); //输出从该点转移到目标的<标签，概率>
                    context.write(outKey,outVal);
                }
            }
        }
        if(fixedDots.contains(name)){
            //固定节点需要将自己的标签转移到自己以保留原来的值
            outKey.set(name);
            for(int i=0;i<dotVal.probabilities.length;i++){
                outVal.set(String.format("%d,%f",i,dotVal.probabilities[i]));
                context.write(outKey,outVal);
            }
        }

        //输出链接列表以维护图结构
        outKey.set(name);
        outVal.set(temp[1]);
        context.write(outKey,outVal);

    }
}
